package com.lyrcsoft.rokectmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @author fujc2dev@126.com
 * @date 2024-05-09 14:56
 */
public class ClusterPushConsumer1 {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("123");
        consumer.setNamesrvAddr("10.20.200.144:9876");
        consumer.setMaxReconsumeTimes(1);
        //设置要读取的topic
        consumer.subscribe("TopicTest",     //指定要读取的消息主题
                "*");                 //指定要读取的消息过滤信息,多个标签数据，则可以输入"tag1 || tag2 || tag3"
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 注册回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String topic = msg.getTopic();
                        String msgBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        String tags = msg.getTags();
                        System.out.println("收到消息：" + "topic：" + topic + "，tags：" + tags + "，body：" + msgBody);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}
